package bsf.baseservice.businessmq.db;


import bsf.baseservice.businessmq.dal.*;
import bsf.baseservice.businessmq.model.*;
import bsf.baseservice.businessmq.producter.*;
import bsf.baseservice.businessmq.systemruntime.*;
import bsf.db.DbConn;

public class ProducterBLL extends BaseBLL
{
	public ProducterInfo getProducterInfo(DbConn PubConn, String mqpath, String productername)
	{
		ProducterInfo r = new ProducterInfo();
		tb_mqpath_dal mqpathdal = new tb_mqpath_dal();
		r.MqPathModel=mqpathdal.get(PubConn, mqpath);
		if (r.MqPathModel == null)
		{
			throw new BusinessMQException(String.format("当前mqpath:%1$s未在MQ中注册队列,请联系管理员注册成功后使用。",mqpath));
		}
		tb_mqpath_partition_dal mqpathpartitiondal = new tb_mqpath_partition_dal();
		r.MqPathParitionModel=mqpathpartitiondal.getList(PubConn, r.MqPathModel.id);
		if (r.MqPathParitionModel == null||r.MqPathParitionModel.isEmpty())
		{
			throw new BusinessMQException(String.format("当前mqpath:%1$s未在MQ中分配相应的分区,请联系管理员分配分区后使用。", mqpath));
		}

		//注册生产者
		r.ProducterModel=this.registerProducter(PubConn, r.MqPathModel.id, productername);
		//获取分区相关节点
		java.util.ArrayList<Integer> datanodepartition = new java.util.ArrayList<Integer>();
		for (tb_mqpath_partition_model p : r.MqPathParitionModel)
		{
			PartitionIDInfo partitionidinfo = PartitionRuleHelper.getPartitionIDInfo(p.partitionid);
			if (!datanodepartition.contains(partitionidinfo.DataNodePartition))
			{
				datanodepartition.add(partitionidinfo.DataNodePartition);
			}
		}
		r.DataNodeModelDic=this.getDataNodeModelsDic(PubConn, datanodepartition);

		return r;
	}

	public tb_producter_model registerProducter(DbConn PubConn, int mqpathid, String productername)
	{
		long tempid = CommonHelper.generateIntID();
		tb_producter_dal dal = new tb_producter_dal();
		dal.deleteNotOnLineByMqpathid(PubConn, mqpathid, SystemParamConfig.Producter_Heartbeat_MAX_TIME_OUT);
		tb_producter_model tempVar = new tb_producter_model();
		tempVar.tempid=tempid;
		tempVar.ip=CommonHelper.getDefaultIP();
		tempVar.mqpathid=mqpathid;
		tempVar.productername=productername;
		dal.add2(PubConn, tempVar);
		return dal.get(PubConn, tempid, mqpathid);
	}

	public void removeProducter(DbConn PubConn, long tempid, int mqpathid)
	{
		tb_producter_dal dal = new tb_producter_dal();
		dal.deleteClient(PubConn, mqpathid, tempid);

	}

	public void producterHeartbeat(DbConn PubConn, long tempid, int mqpathid)
	{
		tb_producter_dal dal = new tb_producter_dal();
		dal.clientHeatbeat(PubConn, mqpathid, tempid);
	}
}